package org.sss.mapreduce.example;

import org.slf4j.Logger;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.datatype.DCPackableBytes;
import org.sss.mapreduce.datatype.PackableBytes;
import org.sss.server.SssServer;

public class DataSpreader extends Reducer {
  private static Logger logger = SssServer.getLogger(DataSpreader.class);
  private int keySize;
  private int keyCard;
  private int keyCount;
  private int valSize;
  private boolean localOnly;
  @Override
  public void configure(Context context) throws Exception {
    Configuration conf = context.getConfiguration();
    this.keySize = conf.getInt("dataspreader.key.size");
    this.keyCard = conf.getInt("dataspreader.key.cardinality");
    this.keyCount = conf.getInt("dataspreader.key.count");
    this.valSize = conf.getInt("dataspreader.val.size");
    this.localOnly = conf.getBoolean("dataspreader.localonly");
    logger.debug("dataspreader.key.size={}", keySize);
    logger.debug("dataspreader.key.cardinarity={}", keyCard);
    logger.debug("dataspreader.key.count={}", keyCount);
    logger.debug("dataspreader.val.size={}", valSize);
    logger.debug("dataspreader.localonly={}", localOnly);
  }
  public void reduce(Context context, DCPackableBytes k, Iterable<PackableBytes> vs, 
    Output<DCPackableBytes, PackableBytes> output) throws Exception {
    int count = 0;
    for (@SuppressWarnings("unused") PackableBytes v: vs) {
      count++;
    }
    logger.debug("k.hashCode={}, k.slot()={}", k.hashCode(), k.slot());
    logger.debug("count={}", count);
    byte[] val = new byte[valSize];
    for (int i=0; i<valSize; i++) {
      val[i] = 0;
    }
    byte[] key = new byte[keySize];
    for (int i=0; i<keySize; i++) {
      key[i] = 0;
    }
    for (int card=0; card<keyCard; card++) {
      key[0] = (byte) ( card        & 0xff);
      key[1] = (byte) ((card >>  8) & 0xff);
      key[2] = (byte) ((card >> 16) & 0xff);
      key[3] = (byte) ((card >> 24) & 0xff);
      if (localOnly) {
        final int node = k.hashCode();
        final int slot = k.slot();
        for (int i=0; i<keyCount; i++) {
          output.write(new DCPackableBytes(key, node, slot),
                       new PackableBytes(val));
        }
      }
      else {
        for (int node=0; node<16; node++) {
          final int slot = k.slot();
          for (int i=0; i<keyCount; i++) {
            output.write(new DCPackableBytes(key, node, slot),
                         new PackableBytes(val));
          }
        }
      } // else
    }
  }
}
